package com.cssl.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cssl.utils.CommonUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;

/**
 * @Author: chen
 * @Date: 2021/11/11 20:35
 * @Desc: 行为数据分流
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //1. 获取执行环境
        Properties properties = CommonUtils.getProperties();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2. 消费ods_base_db主题数据创建流
        String topic = properties.getProperty("kafka.topic.log.ods");
        DataStreamSource<String> kafkaDS = env.addSource(CommonUtils.getKafkaConsumer(topic));

        //3. 将每行数据转换为json格式
        OutputTag<String> dirtyOutputTag = new OutputTag<String>("Dirty") {
        };
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context context, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    collector.collect(jsonObject);
                } catch (Exception e) {
                    //发生异常,将脏数据写入侧输出流
                    context.output(dirtyOutputTag, value);
                }
            }
        });
        //打印脏数据
        jsonObjDS.getSideOutput(dirtyOutputTag).print("dirty>>>>>");

        //4. 新老用户校验 状态编程
        SingleOutputStreamOperator<JSONObject> jsonObjWithNewFlag = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"))
                .map(new RichMapFunction<JSONObject, JSONObject>() {
                    private ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("value-state", String.class));
                    }

                    @Override
                    public JSONObject map(JSONObject value) throws Exception {
                        //获取数据中的 is_new 标记
                        String isNew = value.getJSONObject("common").getString("is_new");
                        if ("1".equals(isNew)) {
                            String state = valueState.value();
                            if (state != null) {
                                //修改 is_new 标记
                                value.getJSONObject("common").put("is_new", "0");
                            } else {
                                valueState.update(isNew);
                            }
                        }
                        return value;
                    }
                });

        //5. 分流
        OutputTag<String> startOutputTag = new OutputTag<String>("start") {
        };
        OutputTag<String> displayOutputTag = new OutputTag<String>("display") {
        };

        SingleOutputStreamOperator<String> pageDS = jsonObjWithNewFlag.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context context, Collector<String> collector) throws Exception {
                String start = value.getString("start");
                if (start != null && start.length() > 0) {
                    //将数据写入启动日志侧输出流
                    context.output(startOutputTag, value.toJSONString());
                } else {
                    //将日志写入页面日志主流
                    collector.collect(value.toJSONString());

                    //取出数据中的爆光数据
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null && displays.size() > 0) {
                        String pageId = value.getJSONObject("page").getString("page_id");
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            //添加页面id
                            display.put("page_id", pageId);
                            //将输出写出到爆光侧输出流
                            context.output(displayOutputTag, display.toJSONString());
                        }
                    }

                }
            }
        });

        //6. 提取侧输出流
        DataStream<String> startDS = pageDS.getSideOutput(startOutputTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayOutputTag);

        //7. 将三个流进行打印并输出到对应的kafka主题中
        String startTopic = properties.getProperty("kafka.topic.log.dwd.start");
        String pageTopic = properties.getProperty("kafka.topic.log.dwd.page");
        String displayTopic = properties.getProperty("kafka.topic.log.dwd.display");

        startDS.addSink(CommonUtils.getKafkaProducer(startTopic));
        pageDS.addSink(CommonUtils.getKafkaProducer(pageTopic));
        displayDS.addSink(CommonUtils.getKafkaProducer(displayTopic));

        //8. 启动任务
        env.execute("BaseLogApp");
    }
}
